#YashanDB Flink Connector

YashanDB连接器允许从YashanDB数据库读取数据和写入数据。

本文档将介绍如何设置YashanDB连接器以对YashanDB数据库运行SQL查询。

# 版本配套说明

# 前提条件

• 已安装Flink，具体操作请查阅Flink官方文档 (opens new window)。

• 已获取下列jar包并存放至<FLINK_HOME>/lib/目录下：

  可自行使用构建自动化工具（例如Maven）将项目构建打包，也可联系我们的技术支持获取。

  • flink-connector-yashandb.jar

  • yashandb-jdbc.jar

# 使用限制

YashanDB flink connector不对时间类型的时区进行处理，如遇时区问题导致时间差异，可参考Flink官网文档相关介绍 (opens new window)。

# 配置YashanDB

为了让连接器能够正常运行，您需要为连接器的连接用户授予相关的权限。

YashanDB CDC connector的连接用户需要相关权限，请授予如下权限保障任务能够正常运行：

GRANT CREATE SESSION TO username;
GRANT SELECT ANY TABLE TO username;
GRANT INSERT ANY TABLE TO username;

# 创建YashanDB表

YashanDB表可以定义如下：

-- register an YashanDB table 'products' in Flink SQL
Flink SQL> CREATE TABLE products (
     ID INT NOT NULL,
     NAME STRING,
     DESCRIPTION STRING,
     WEIGHT DECIMAL(10, 3),
     PRIMARY KEY(id) NOT ENFORCED
     ) WITH (
     'connector' = 'yashandb',
     'url' = 'jdbc:yasdb://192.168.3.216:1665/test',
     'port' = '1688',
     'username' = 'flinkuser',
     'password' = 'flinkpw',
     'schema-name' = 'inventory',
     'table-name' = 'products');
  
-- 查看YashanDB表中的数据
Flink SQL> SELECT * FROM products;

# 连接器参数

# 功能特性

# 键处理

当写入数据到外部数据库时，Flink会使用DDL中定义的主键。如果定义了主键，则连接器将以upsert模式工作，否则连接器将以append模式工作。

• 在upsert模式下，connector将根据主键判断插入新行或更新已存在的行，该模式可确保幂等性。建议为表定义主键并确保该主键为底层数据库中表的唯一键或主键。

• 在append模式下，connector会把所有记录解释为INSERT消息，如果违反了底层数据库中主键或者唯一约束，INSERT插入可能会失败。

PRIMARY KEY语法的更多详情，请查阅Flink官方文档 (opens new window)。

# 分区扫描

分区扫描相关介绍，请查阅Flink官方文档 (opens new window)。

# 幂等写入

如果在DDL中定义了主键，JDBC sink将使用upsert语义（而非普通的INSERT语句），如果底层数据库中存在违反唯一性约束，则原子地添加新行或更新现有行，从而确保幂等性。

如果出现故障，Flink作业会从上次成功的checkpoint恢复并重新处理，基于该机制可能会在恢复过程中重复处理消息。如果使用upsert模式，在需要重复处理记录的场景中可有效避免违反数据库主键约束和产生重复数据。

除了故障恢复场景外，数据源（kafka topic）也可能随着时间推移自然地包含多个具有相同主键的记录，因此upsert模式将更符合预期。

# BulkLoad模式

为提高YashanDB的LSC表的写入性能，本连接器提供bulkload模式，可通过bulk-load-enable设置，若开启bulkload模式（即设置为true），原普通插入将变成bulkInsert、原upsert插入将变成bulkupsert，使用BULKLOAD模式的详细介绍请查阅hint。

# 数据类型映射